package com.lzq.mq.common.event;

import com.lzq.mq.common.utils.ReflectUtils;
import com.lzq.mq.common.event.model.Event;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 事件总线
 */
@Slf4j
public class EventBus {

    private final ThreadPoolExecutor threadPoolExecutor;
    /**
     * 监听器订阅关系
     * key: 监听器所监听的事件类型
     * value: 监听器
     */
    // TODO 多线程问题
    // TODO 重复初始化
    private static final Map<Class<? extends Event>, List<Listener>> listenerMap = new HashMap<>();

    public EventBus(String name) {
        threadPoolExecutor = new ThreadPoolExecutor(
                10,
                100,
                3,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(1000),
                r -> {
                    Thread thread = new Thread(r);
                    thread.setName(name + thread.getId());
                    return thread;
                }
        );
    }

    public EventBus() {
        this("event-bus-thread-");
    }

    public static void init() {

        ServiceLoader<Listener> listeners = ServiceLoader.load(Listener.class);
        for (Listener listener : listeners) {
            Class clazz = ReflectUtils.getInterfaceT(listener,0);
            registry(clazz, listener);
            log.info("注册数据总线监听器:{}->{}", clazz.getName(), listener.getClass().getName());
        }
    }

    /**
     * 注册监听器
     * @param eventClass 事件类型
     * @param listener 监听器
     */
    private static void registry(Class<? extends Event> eventClass, Listener listener) {

        List<Listener> listeners = listenerMap.get(eventClass);

        if (listeners == null) {
            listeners = new ArrayList<>();
            listenerMap.put(eventClass, listeners);
        }
        listeners.add(listener);
    }

    /**
     * 发送事件
     * @param event 事件对象
     */
    public void publish(Event event) {
        List<Listener> listeners = listenerMap.get(event.getClass());
        if (listeners == null) {
            return;
        }
        threadPoolExecutor.execute(() -> {
            try {
                for (Listener listener : listeners) {
                    listener.onReceive(event);
                }
            } catch (Exception e) {
                log.error("数据总线发送消息出错:", e);
            }
        });
    }
}
